package com.zhny.test;

import org.apache.parquet.Strings;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
// $example on$
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Arrays;

import org.apache.spark.api.java.JavaDoubleRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.mllib.linalg.Matrix;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.mllib.stat.Statistics;
//皮尔森
public class PearsonUtil {
    public static void exc(String dataFilePath, String resultFilePath) {
        SparkConf conf = new SparkConf().setAppName("JavaCorrelationsExample").setMaster("local[*]");
        JavaSparkContext jsc = new JavaSparkContext(conf);

        // $example on$
//        JavaDoubleRDD seriesX = jsc.parallelizeDoubles(
//                Arrays.asList(1.0, 2.0, 3.0, 3.0, 5.0));  // a series

        // must have the same number of partitions and cardinality as seriesX
//        JavaDoubleRDD seriesY = jsc.parallelizeDoubles(
//                Arrays.asList(11.0, 22.0, 33.0, 33.0, 555.0));

        // compute the correlation using Pearson's method. Enter "spearman" for Spearman's method.
        // If a method is not specified, Pearson's method will be used by default.
//        double correlation = Statistics.corr(seriesX.srdd(), seriesY.srdd(), "pearson");
//        System.out.println("Correlation is: " + correlation);

        // note that each Vector is a row and not a column

        JavaRDD<Vector> data = null;

        if (Strings.isNullOrEmpty(dataFilePath)) {
            data = jsc.parallelize(
                Arrays.asList(
                        Vectors.dense(1.0, 10.0, 100.0),
                        Vectors.dense(2.0, 20.0, 200.0),
                        Vectors.dense(5.0, 33.0, 366.0)
                )
            );
        } else {
            JavaRDD<String> dataFile = jsc.textFile(dataFilePath);

            data = dataFile.map(new Function<String, Vector>() {
                private static final long serialVersionUID = 1L;

                @Override
                public Vector call(String line) throws Exception {
                    String[] elements = line.split(",");

                    double[] vectors = new double[elements.length];

                    for (int i = 0; i < elements.length; i++) {
                        vectors[i] = Double.parseDouble(elements[i]);
                    }

                    return Vectors.dense(vectors);
                }
            }).cache();
        }

        // calculate the correlation matrix using Pearson's method.
        // Use "spearman" for Spearman's method.
        // If a method is not specified, Pearson's method will be used by default.
        Matrix correlMatrix = Statistics.corr(data.rdd(), "pearson");
        // $example off$
        FileWriter fos = null;

        try {
            fos = new FileWriter(new File(resultFilePath));
            fos.write(correlMatrix.toString());

            fos.flush();
            fos.close();
        } catch (IOException e) {
            e.printStackTrace();
        }

        jsc.stop();
    }
}
